package com.zengjianjun.storm.demo.wordcount;

import lombok.extern.slf4j.Slf4j;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Map;

/**
 * 统计数量
 *
 * @author zengjianjun
 */
@Slf4j
public class CountBolt extends AbstractBaseRichBolt {

    private final Map<String, Integer> countMap = new HashMap<>();

    @Override
    public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {

    }

    @Override
    public void execute(Tuple input) {
        String word = input.getStringByField("word");
        Integer count = countMap.get(word);
        if (count == null) {
            count = 0;
        }
        count++;
        countMap.put(word, count);
        // 输出
        log.info("当前实时统计结果: {}:{}", word, count);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}
